package com.atguigu.flink.chapter05.transform;

import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author lzc
 * @Date 2023/6/19 08:52
 */
public class RichFunctionDemo {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
        // 查询数据库中的某张表, 如果存在这个值, 则过滤掉, 否则就保留下来来
    
        env
        
            //.fromElements(10, 21, 31, 41)  // etl
            .socketTextStream("hadoop162", 8888)
            .map(x -> 1)
            .map(new RichMapFunction<Integer, Integer>() {
                @Override
                public void open(Configuration parameters) throws Exception {
                    System.out.println("RichFunctionDemo.open.. map");
                }
            
                @Override
                public Integer map(Integer value) throws Exception {
                    System.out.println(value + "map");
                    return value;
                }
            })
            .filter(new RichFilterFunction<Integer>() {
                @Override
                public void open(Configuration parameters) throws Exception {
                    // 建立到 mysql 的链接
                    System.out.println("RichFunctionDemo.open... filter");
                }
            
                @Override
                public void close() throws Exception {
                    // 关闭链接
                    System.out.println("RichFunctionDemo.close");
                }
            
                @Override
                public boolean filter(Integer value) throws Exception {
                    System.out.println(value);
                    // 使用链接
                    return true;
                }
            
            })
            .print("a");
           
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
/*
reich.. 函数:
    1. 多了两个生命周期方法
        open:
            程序启动的时候, 首先执行.
            一个并行度只执行一次
        close
            程序正常停止的时候执行,
            一个并行度执行一次
 */